#pragma once  
  
extern "C"
{
#include "./MQTTAsync.h" 
}
 
#include <map>
#include <vector>

#include "./threadLock.h"

#include "./fore_rtdb.h"
#include "./base.h"
 
#include "./rtdbFun.h"
 

#include "./mqtt_msg.h"


#define PUB_CLIENTID    "MY_Pub_CLIENTID"
#define thread_type pthread_t
#define thread_return_type void*
 

#define QOS         1
 
typedef std::vector<int> ARYINT;

#pragma pack(1)

class CMQTTAsyncPubCli;
 
struct pub_thread_parms
{
	MQTTAsync c; 
	int qos;
	int index; 
	FORE_DEV_TYPE  devTypeInfo;  
	CMQTTAsyncPubCli * pPubCli;
};

struct THREAD_STATUS
{
	thread_type   handle;
	unsigned char status;
};

#pragma pack()

/*
#ifdef WIN32
#include "windows.h"
typedef thread_return_type (*thread_fn)(void*);
inline void * Thread_start(thread_fn fn, void* parameter)
{
 
	DWORD ThreadId = 0; 
  
	HANDLE handle = CreateThread(   NULL,
									0,
									(LPTHREAD_START_ROUTINE)fn,
									(LPVOID)parameter,
									0,
									&ThreadId );

	if( handle == NULL )
	{ 
	}

	return handle;
}



	 
#else 
#define LPTHREAD_START_ROUTINE   void*(*)(void*)
inline thread_type Thread_start ( 
		  void* (*lpStartAddress)(void*)  ,
		  void*  ThreadParameter 
		)
{                    
             
	pthread_t   threadId;
	pthread_attr_t attr;

	pthread_attr_init(&attr);
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

	pthread_create(&threadId, &attr, (void*(*)(void*))lpStartAddress, ThreadParameter);


	pthread_attr_destroy(&attr); 
	
	return threadId;
}

#endif*/
 

class CMQTTAsyncPubCli  
{  
public:  
    CMQTTAsyncPubCli();  
    ~CMQTTAsyncPubCli();  
	
	
	bool   connect_mqtt(char * mqtt_svr_addr);
	bool   re_connect_mqtt();
	void   disconnect(); 
	void CloseThreadHandle(char * devType = NULL);
	void create_devTypePubtopic_thread(char * devType = NULL);
	bool ReStartPubThreasd(  ) ;
	bool on_subscribe();
private:
	MQTTAsync 					m_client;
	 
	string						m_mqtt_svr_addr;
	
	
 
public:
	#ifdef WIN32
		std::map<string, HANDLE>					m_aryDevTypeThreadHandle;
	#else
		std::map<string, THREAD_STATUS>				m_aryDevTypeThreadHandle;  
	#endif
	
};


#define ADDRESS     "tcp://127.0.0.1:1883"
#define SUB_CLIENTID    "Sub_ClientSVR"


struct STRU_UPDATE_RTDB
{
	char    serviceId[32];
	char    field_name[32];
	char	val[32];
	char    quality[12];
};

struct QueryRtdb
{
	char dev_code[64];
	 
	 std::vector<string>  aryfieldName;
};

struct ADDR_REG
{
	char    addr[13];
	char	devicesid[64];
};

struct QUERY_MODEL_DEV
{
	char    query_FieldName[32];
	char	query_FieldVal[32];
};

class CMQTTAsyncSubCli  
{  
public:  
    CMQTTAsyncSubCli();  
    ~CMQTTAsyncSubCli();  
	
	 
	bool   connect_mqtt(char * client_id, char * mqtt_svr_addr ) ;
	bool   re_connect_mqtt();

	bool   on_subscribe(); 

	MQTTAsync 					m_client;

	CMQTTAsyncPubCli			* m_pPubCli;

	bool					m_bReInitRtdb;

public:
	bool terminal_reg(char* topic, char* msg);				//�ն�ע�� 
	bool _analyse_terminialReg_msg(char* jsonMsg, string & strToken, FORE_DEV_DICT &  devInfo);  //�ն�ע��(485��߼�app)
	bool Create_MeaInfo(FORE_DEV_DICT &  devInfo);

public:
	bool query_model(char* appName, char* msg);  //����ģ�Ͳ�ѯ�ӿ�
	bool query_devModel(char* appName, char* msg);  //�豸ģ�Ͳ�ѯ�ӿ�

	bool query_param(char* appName, char* msg);  //��ֵ��ѯ�ӿ�
	bool set_param(char* appName, char* msg);  //��ֵ���ýӿ�
	bool remove_param(char* appName, char* msg);  //��ֵ���ýӿ�

	bool update_rtdb(char* dev_type_device_code, char* msg);  //���������ϱ��ӿ�
	//bool query_relationship(char* appName, char* msg);  //��ϵ��ѯ�ӿ�
	bool query_rtdb(char* appName, char* msg);			//��ѯʵʱ����

	bool reg_subscribe_topic(char* appName, char* msg);			//ע�ᶩ������
	 
	bool app_reg(char* appName, char* msg);				//APPע��
	//bool addr_reg(char* appName, char* msg);			//�豸��ַע��

	bool query_DevType(char* appName, char* jsonMsg);

	

	bool query_terminal(char* appName, char* jsonMsg);		//��ѯ��ע����ն�

	//�豸ģ�Ͳ����ӿ�
	/*bool insert_model_dev_info(char* appName, char* jsonMsg);		//����ģ���豸��Ϣ
	bool insert_model_param(char* appName, char* jsonMsg);			//����ģ�Ͳ�����Ϣ
	bool insert_model_analog(char* appName, char* jsonMsg);			//����ģ���豸��analog��Ϣ
	bool insert_model_discrete(char* appName, char* jsonMsg);		//����ģ���豸��discrete��Ϣ
	bool insert_model_accumulator(char* appName, char* jsonMsg);		//����ģ���豸��accumulator��Ϣ
	bool insert_model_command(char* appName, char* jsonMsg);		//����ģ���豸��command��Ϣ
	bool remove_model(char* appName, char* jsonMsg);			//ɾ��ģ���豸 */
	bool query_model_dev_byDevType(char* appName, char* jsonMsg);			//��ѯ�豸ģ����Ϣ
	//bool query_model_parameter(char* appName, char* jsonMsg);		//��ѯ�豸����ģ����Ϣ
	//bool query_model_measurement(char* appName, char* jsonMsg);		//��ѯ�豸measurementģ����Ϣ
private:
	//����ģ�Ͳ�ѯ�ӿ�
	bool _analyse_query_model_msg(char* jsonMsg, string & strToken, std::vector<string> & aryDevType);   
	bool _process_model_fun(char* appName, string & strToken, std::vector<string> & aryDevType);
	 
	//��ֵ��ѯ�ӿ�
	bool _analyse_query_param_msg(char* jsonMsg, string & strToken,  std::vector<string> &  aryParam);
	bool _process_parameter_fun(char* appName, string & strToken, std::vector<string> & aryParam);

	//���������ϱ��ӿ�
	bool _analyse_update_rtdb_msg(char* jsonMsg, string & strToken, string & strTime,  std::vector<STRU_UPDATE_RTDB> &  aryUpdateRtdb);
	bool _process_updatertdb_fun(FORE_DEV_DICT * pDev, char * appName, string & strTime, std::vector<STRU_UPDATE_RTDB> &  aryUpdateRtdb);
	bool auto_upgrade_serviceId(FORE_DEV_DICT * pDev, STRU_UPDATE_RTDB & rtdb_data);

	//��ϵ��ѯ�ӿ�
	//bool _process_relationship_fun(char* appName, string & strToken, std::vector<string> & aryDevType);
	
	//��ѯʵʱ����
	bool _analyse_query_rtdb_msg(char* jsonMsg, string & strToken,  std::vector<QueryRtdb> &  aryQueryRtdb, char * chrError);
	bool _process_query_rtdb_fun(char* appName, string & strToken, std::vector<QueryRtdb> & aryQueryRtdb, char * chrError);

	//APPע��
	//bool _analyse_app_reg_msg(char* jsonMsg, string & strToken,  FORE_APP_REG &  appReg);
	//bool _process_app_reg_fun(char* appName, string & strToken, FORE_APP_REG &  appReg);

	//�豸��ַע��
	/*bool _analyse_addr_reg_msg(char* jsonMsg, string & strToken,  std::vector<ADDR_REG> &  AddrReg);
	bool _process_addr_reg_fun(char* appName, string & strToken, std::vector<ADDR_REG> &  AddrReg);*/

	//�豸ģ�Ͳ�ѯ�ӿ�
	bool _process_devModel_fun(char* appName, string & strToken, std::vector<string> & aryDevType);

	//��ֵ�趨�ӿ�
	bool _analyse_set_param_msg(char* jsonMsg, string & strToken,  std::vector<FORE_PARAM_CFG> &  aryParam);
	bool _process_set_param_fun(char* appName, string & strToken, std::vector<FORE_PARAM_CFG> & aryParam);

	//����ɾ��
	bool _process_remove_param_fun(char* appName, string & strToken, std::vector<string> & aryParam);

	//ע�ᶩ������
	//bool _analyse_subscribTopic_msg(char* appName, char* jsonMsg, string & strToken,  std::vector<FORE_APP_SUBSCRIBE_REG> & arySubscribeReg);
	//bool _process_subscribeTopic_fun(char* appName, string & strToken,  std::vector<FORE_APP_SUBSCRIBE_REG> & arySubscribeReg);

	

	//��ѯ��ע����ն� 
	bool _analyse_queryDevReg_msg(char * appName, char* jsonMsg, string & strToken, string & strManufacturerId, string & strDeviceType, string & strModel);
	bool _queryDevReg_from_rtdb(char * appName, string & strToken, string & strManufacturerId, string & strDeviceType, string & strModel);


	//����ģ���豸��Ϣ
	bool _analyse_insert_dev_msg( char* jsonMsg, string & strToken );
	bool _analyse_insert_param_msg( char* jsonMsg, string & strToken );
	bool _analyse_insert_analog_msg( char* jsonMsg, string & strToken );
	bool _analyse_insert_discrete_msg( char* jsonMsg, string & strToken );
	bool _analyse_insert_accumulator_msg( char* jsonMsg, string & strToken );
	bool _analyse_insert_command_msg( char* jsonMsg, string & strToken );
	bool _analyse_remove_devModel_msg( char* jsonMsg, string & strToken, string & strSN, std::vector<string> & aryRemoveNode );

	bool _process_remove_devMode_fun(string & strSN, std::vector<string> & aryRemoveNode);
    	bool _analyse_query_model_dev_msg(char* jsonMsg, string & strToken, std::vector<string> & aryDevType, char * chrError); 
	bool _query_model_dev_from_db(char * appName, string &strToken, std::vector<string> & aryDevType, char * chrError);

	bool _analyse_query_model_param_msg(char* jsonMsg, string & strToken, string & strSN, std::vector<string> & aryParamName);
	bool _query_model_param_from_db(char* appName, string & strToken, string & strSN, std::vector<string> & aryParamName);
	
	bool _analyse_query_model_measurementr_msg(char* jsonMsg, string & strToken, string & strSN, std::vector<string> & aryMeaName, char * chrError);
	bool _query_model_measurementr_from_db(char* appName, string & strToken, string & strSN, std::vector<string> & aryMeaName, char * chrError);

	bool _analyse_query_devtype_msg(char* jsonMsg, string & strToken,  std::vector<string> &  aryDevtype);
	bool _process_devtype_fun(char * appName, string &strToken, std::vector<string> & aryDevType );
private:
	
	//CThreadLock					m_appreg_threadLock; 

	string						m_mqtt_svr_addr;
	string						m_client_sub_id;
	
	void disconnect(); 
	bool Send_Reply_pack(char * topip, char * chrToken, int statusCode, char* statusDesc,  char * errorDesc=NULL, char* errorTag=NULL);
	bool  _send_string( char* pTopic, unsigned char pub_dformat_type, char * msg_data );
	
};

 


